package com.teradata.config;

import com.teradata.callback.CallBack;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @Project: edb
 * @Description:
 * @Version 1.0
 * @Throws :
 * @Author: <li>2019/4/18 lyndon Create 1.0
 * @Copyright (c)2018-2018 中央结算公司，版权所有
 * @Modified By:
 */
public class TaskExecutor {

    private final BlockingQueue<TaskContext> monitorQueue = new LinkedBlockingQueue<TaskContext>();

    private final Thread monitorThread;

    private ThreadPoolTaskExecutor pool;

    public TaskExecutor(ThreadPoolTaskExecutor pool) {
        this.pool = pool;
        this.monitorThread = new Thread(() -> {
            while (true) {
                dispatch();
            }
        }, "async-monitor");

        this.monitorThread.setDaemon(true);
        this.monitorThread.start();
    }

    public void monitor(TaskContext taskContext) {
        try {
            monitorQueue.put(taskContext);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private void dispatch() {
        try {
            final TaskContext taskContext = monitorQueue.poll(120, TimeUnit.SECONDS);

            if (taskContext == null) {
                return;
            }

            if (!taskContext.empty() && taskContext.lock()) {
                taskContext.setDispatchTime(System.currentTimeMillis());
                try {
                    this.pool.execute(() -> {
                        CallBack cb = null;
                        try {
                            Object[] msg = taskContext.poll();
                            if (msg != null) {
                                cb = msg.length == 2 ? (CallBack) msg[1] : null;

                                if (cb != null) {
                                    cb.complete(true, null);
                                }
                            }
                        } catch (Throwable e) {
                            e.printStackTrace();
                            if (cb != null) {
                                try {
                                    cb.complete(false, e);
                                } catch (Throwable t) {
                                    t.printStackTrace();
                                }
                            }
                        } finally {
                            taskContext.unlock();
                            if (!taskContext.empty()) {
                                monitor(taskContext);
                            }
                        }
                    });
                } catch (Exception e) {
                    e.printStackTrace();
                    taskContext.unlock();
                }
            }
        } catch  (Exception f) {
            f.printStackTrace();
        }
    }


    public void shutdown() {
        this.monitorThread.interrupt();
    }
}
